package cn.xqapp.GatewayClient;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class GatewayClient implements Runnable {

    protected String host;
    protected int port;

    protected int healthCheckInterval;

    protected int clientNum;
    protected List<GatewayClientHandler> clientHandlerList;

    protected int messageSize;
    protected int messageInterval;

    protected ChannelHandler clientChannel;

    public GatewayClient(String host, int port, int clientNum, int messageSize, int messageInterval) {
        this.host = host;
        this.port = port;

        this.healthCheckInterval = 10;

        this.clientNum = clientNum;
        this.clientHandlerList = new ArrayList<>();

        this.messageSize = messageSize;
        this.messageInterval = messageInterval;
    }

    @Override
    public void run() {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // new client list
            for(int i=0; i<this.clientNum; i++){
                Bootstrap b = new Bootstrap();
                b.group(workerGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.SO_KEEPALIVE, true);

                // client handler
                GatewayClientHandler clientHandler = new GatewayClientHandler();

                b.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(
                                new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
                                new LengthFieldPrepender(4),
                                clientHandler);
                    }
                });

                ChannelFuture f = b.connect(this.host, this.port).sync();
                if(f.isDone()){
                    if(f.isSuccess()){
                        clientHandlerList.add(clientHandler);
                    }else{
                        System.out.println(f.cause().getMessage());
                    }
                }
            }

            System.out.println(String.format("%d client connected...", clientHandlerList.size()));

            // start echo message
            byte[] message = new byte[this.messageSize];
            for(GatewayClientHandler clientHandler:clientHandlerList){
                clientHandler.start(message, true, this.messageInterval);
            }

            // start health checker
            ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
            try{
                scheduler.scheduleAtFixedRate(this::checkHealth, 0, this.healthCheckInterval, TimeUnit.SECONDS);

                Scanner scanner = new Scanner(System.in);
                scanner.nextLine();
            }finally {
                scheduler.shutdown();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    protected void checkHealth(){
        try{
            int conn = 0;
            int err = 0;
            long totalMsg = 0;
            long totalTimeNano = 0;
            float totalSpanMS = 0.0f;
            float ops = 0.0f;
            float lag = 0.0f;

            for(GatewayClientHandler clientHandler:clientHandlerList){
                if(clientHandler.isActive()){
                    conn += 1;
                    totalMsg += clientHandler.getSendMessageCount();
                    totalTimeNano += clientHandler.getSendMessageNanoTime();
                    totalSpanMS += clientHandler.getSendSpanMilliseconds();
                }else{
                    err += 1;
                }
            }

            if(totalSpanMS > 0){
                ops = totalMsg * 1000L / totalSpanMS;
            }
            if(totalMsg > 0){
                lag = totalTimeNano / 1000000f / totalMsg;
            }

            System.out.println(String.format("[%s] CON %d MSG %d TIME %d OPS %.2f LAG %.3f ERR %d",
                    "CLIENT", conn, totalMsg, this.healthCheckInterval, ops, lag, err));

            for(GatewayClientHandler clientHandler:clientHandlerList){
                if(clientHandler.isActive()) {
                    clientHandler.reset();
                }
            }
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }
}